package com.example.stomp.websocket.client;

import com.example.stomp.websocket.bean.ChatInfo;
import com.example.stomp.websocket.bean.Greeting;
import com.example.stomp.websocket.bean.HelloMessage;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutionException;

/**
 * @author ZhenWuWang
 */
public class StompUserClient1
{
    private WebSocketStompClient stompClient = null;

    private StompSession stompSession = null;

    private SockJsClient sockJsClient = null;

    private ThreadPoolTaskScheduler task = null;

    private WebSocketHttpHeaders headers = new WebSocketHttpHeaders();//请求头

    public volatile boolean recvFlag = false;


    public static void main(String[] args)
    {
        String sendMsg = "我是xiaoming";

        sendMsg = (args != null && args.length != 0) ? args[0] : sendMsg;

        StompUserClient1 myClient = new StompUserClient1();
        String url = "ws://localhost:8080/chat?name=xiaoming";
        String subscribe = "/user/xiaoming/message";
        String send = "/message";
//        String send = "/topic/greetings";
        try
        {
            myClient.runStompClient(myClient, url, subscribe, send, sendMsg);
        } catch (ExecutionException | InterruptedException | UnsupportedEncodingException | JsonProcessingException e)
        {
            e.printStackTrace();
        }

        while (!myClient.recvFlag)
        {
            //持续等待返回标志位为true
            System.out.println("-------------------持续等待返回验证消息中……，当前flag:" + myClient.recvFlag);
            try
            {
                Thread.sleep(3000);
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
        //关闭所有连接终止程序
//        myClient.task.destroy();
//        myClient.sockJsClient.stop();
//        myClient.stompClient.stop();
//        myClient.stompSession.disconnect();
//        System.exit(0);
    }

    public void runStompClient(StompUserClient1 client, String URI, String subscribe, String send, final String sendMsg)
            throws ExecutionException, InterruptedException, UnsupportedEncodingException, JsonProcessingException
    {
        //连接到对应的endpoint点上，也就是建立起websocket连接
        ListenableFuture<StompSession> f = client.connect(URI);
        //建立成功后返回一个stomp协议的会话
        StompSession stompSession = f.get();

        System.out.println("Subscribing to greeting topic using session " + stompSession);
        //绑定订阅的消息地址subscribe
        client.subscribeGreetings(subscribe, stompSession);
        //设置Receipt头，不设置无法接受返回消息
        stompSession.setAutoReceipt(true);
//        Thread.sleep(5000);
        //绑定发送的的地址send，注意这里使用的字节方式发送数据
        StompSession.Receiptable rec = stompSession.send(send, new ChatInfo("xiaohong",sendMsg));
        //添加消息发送成功的回调
        rec.addReceiptLostTask(() -> System.out.println("消息发送成功,发送内容为:" + sendMsg));
    }

    public ListenableFuture<StompSession> connect(String url)
    {
        Transport transport = new WebSocketTransport(new StandardWebSocketClient());
        List<Transport> transports = Collections.singletonList(transport);
        sockJsClient = new SockJsClient(transports);

//        sockJsClient.setMessageCodec(new Jackson2SockJsMessageCodec());
        stompClient = new WebSocketStompClient(sockJsClient);
        MappingJackson2MessageConverter m = new MappingJackson2MessageConverter();
        stompClient.setMessageConverter(m);

        stompClient.setReceiptTimeLimit(300);
        stompClient.setDefaultHeartbeat(new long[]{100001, 100001});

        task = new ThreadPoolTaskScheduler();
        task.initialize();

        stompClient.setTaskScheduler(task);

        return stompClient.connect(url, headers, new MyHandler(), "localhost", 8080);
    }

    public void subscribeGreetings(String url, StompSession stompSession)
    {
        stompSession.subscribe(url, new StompFrameHandler()
        {
            @Override
            public Type getPayloadType(StompHeaders stompHeaders)
            {
                return ChatInfo.class;
            }

            public void handleFrame(StompHeaders stompHeaders, Object o)
            {
                ChatInfo chatInfo = (ChatInfo) o;
                System.out.println("收到返回的消息" + chatInfo.toString());
                recvFlag = true;
            }
        });
    }

    private class MyHandler extends StompSessionHandlerAdapter
    {
        @Override
        public void handleFrame(StompHeaders headers, Object payload)
        {
            System.out.println("=========================handleFrame");
            super.handleFrame(headers, payload);
        }

        @Override
        public void afterConnected(StompSession session, StompHeaders connectedHeaders)
        {
            System.out.println("连接成功");
            stompSession = session;
        }

        @Override
        public void handleTransportError(StompSession session, Throwable exception)
        {
            System.out.println("连接出现异常");
            exception.printStackTrace();
        }
    }
}
